package transform;

import bean.SensorReading;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/3 10:08
 */
public class Transform2_Rolling {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String inputPath = "src/main/resources/sensor.txt";

        DataStream<String> inputStream = env.readTextFile(inputPath);

        /****
         * 转换为SensorReading类型
         */

/*        DataStream<SensorReading> dataStream =inputStream.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String s) throws Exception {
                String[] fields =s.split(",");
                return new SensorReading(fields[0],new Long(fields[1]),new Double(fields[2]));
            }
        });*/

        /**
         *
         */
        DataStream<SensorReading> dataStream = inputStream.map(line->{
            String[] fields = line.split(",");
            return new SensorReading(fields[0],new Long(fields[1]),new Double(fields[2]));
        });
        /***
         * 分组
         */

        KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id");

        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(data -> data.getId());
        KeyedStream<SensorReading, String> keyedStream2 = dataStream.keyBy(SensorReading::getId);


        /**
         * 聚合
         */

        /**
         * max，取最大值的 初始时间戳
         *max> SensorReading{id='sensor_1', timestamp=1547718199, temperature=35.8}
         * max> SensorReading{id='sensor_6', timestamp=1547718201, temperature=15.4}
         * max> SensorReading{id='sensor_7', timestamp=1547718202, temperature=6.7}
         * max> SensorReading{id='sensor_10', timestamp=1547712205, temperature=66.6}
         * max> SensorReading{id='sensor_1', timestamp=1547718199, temperature=36.8}
         * max> SensorReading{id='sensor_1', timestamp=1547718199, temperature=36.8}
         * max> SensorReading{id='sensor_1', timestamp=1547718199, temperature=37.8}
         * max> SensorReading{id='sensor_1', timestamp=1547718199, temperature=38.8}
         */
        DataStream<SensorReading> resultStream = keyedStream.max("temperature");
        /**
         * maxBy，取最大值 位于 的时间戳
         * max> SensorReading{id='sensor_1', timestamp=1547718199, temperature=35.8}
         * max> SensorReading{id='sensor_6', timestamp=1547718201, temperature=15.4}
         * max> SensorReading{id='sensor_7', timestamp=1547718202, temperature=6.7}
         * max> SensorReading{id='sensor_10', timestamp=1547712205, temperature=66.6}
         * max> SensorReading{id='sensor_1', timestamp=1547718299, temperature=36.8}
         * max> SensorReading{id='sensor_1', timestamp=1547718299, temperature=36.8}
         * max> SensorReading{id='sensor_1', timestamp=1547718499, temperature=37.8}
         * max> SensorReading{id='sensor_1', timestamp=1547718599, temperature=38.8}
         */
        DataStream<SensorReading> resultStream1 = keyedStream.maxBy("temperature");

        resultStream.print("max");
        env.execute("");


    }
}
